package com.yihaodian.cluster.discovery.zk;

import java.util.Collection;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;

import org.apache.log4j.Logger;

import com.google.common.collect.Lists;
import com.google.gson.Gson;
import com.google.inject.Inject;
import com.google.inject.name.Named;
import com.yihaodian.api.common.json.GsonWrapper;
import com.yihaodian.cluster.Cluster;
import com.yihaodian.cluster.ClusterState;
import com.yihaodian.cluster.Node;
import com.yihaodian.cluster.Node.Role;
import com.yihaodian.cluster.discovery.NodeDiscovery;
import com.yihaodian.server.AbstractLifecycleService;
import com.yihaodian.zk.AbstractNodeListener;
import com.yihaodian.zk.ZkClient;
import com.yihaodian.zk.ZkClientSessionExpiredException;

public class ZkNodeDiscovery extends AbstractLifecycleService implements
		NodeDiscovery {
	public static boolean clearState = false;

	private static final Logger logger = Logger
			.getLogger(ZkNodeDiscovery.class);

	private volatile Thread currentJoinThread;

	private final Lock updateNodeListLock = new ReentrantLock();

	Gson gson;

	ZkClient zkClient;

	ZkPath zkPath;

	Node localNode;

	CopyOnWriteArrayList<NewClusterStateListener> newClusterStateListeners = new CopyOnWriteArrayList<NewClusterStateListener>();
	CopyOnWriteArrayList<NodeChangeListener> nodeChangeListeners = new CopyOnWriteArrayList<NodeChangeListener>();
	CopyOnWriteArrayList<InitClusterStateListener> initClusterStateListeners = new CopyOnWriteArrayList<InitClusterStateListener>();

	@Inject
	public ZkNodeDiscovery(Node localNode, ZkClient zkClient,
			@Named("zkRootPath") String zkRootPath) {
		this.zkClient = zkClient;
		this.localNode = localNode;
		this.zkPath = new ZkPath(zkRootPath, this.localNode.getClusterId());
		this.gson = GsonWrapper.getGson();
	}

	@Override
	public void addListener(NewClusterStateListener listener) {
		newClusterStateListeners.add(listener);
	}

	@Override
	public void addListener(NodeChangeListener listener) {
		nodeChangeListeners.add(listener);
	}

	@Override
	public void addListener(InitClusterStateListener listener) {
		initClusterStateListeners.add(listener);
	}

	@Override
	public Node getLocalNode() {
		return localNode;
	}

	/**
	 * publish/broadcast cluster state to ZK
	 */
	@Override
	public void publishState(ClusterState clusterState) {
		if (localNode.getRole() != Role.MASTER) {
			logger.warn(localNode.getId()
					+ " is not master, can not publish state ");
			// return;
		}

		try {

			byte[] masterNode = zkClient.getNode(zkPath.getMasterPath(), null);
			if (masterNode == null
					|| !new String(masterNode).equals(localNode.getId())) {
				logger.warn("No longer a master, shouldn't publish new state");
				return;
			}

			/**
			 * gson.toJson(clusterState) has error
			 */
			String json = gson.toJsonTree(clusterState, ClusterState.class)
					.toString();
			logger.info("publish state: " + json);
			zkClient.setOrCreatePersistentNode(zkPath.getStatePath(),
					json.getBytes());
			// TODO, large node
			// zkClient.createLargeSequentialNode();
		} catch (Exception e) {
			logger.error("publish state error ", e);
		}
	}

	/**
	 * retrieve cluster state from zk, watch the zk node.
	 * 
	 * @param newClusterStateListener
	 * @return
	 * @throws InterruptedException
	 */
	public ClusterState retrieveState(
			final NewClusterStateListener newClusterStateListener)
			throws InterruptedException {
		ZkClient.NodeListener nodeListener;
		if (newClusterStateListener != null) {
			nodeListener = new AbstractNodeListener() {
				@Override
				public void onNodeCreated(String id) {
					updateClusterState(newClusterStateListener);
				}

				@Override
				public void onNodeDataChanged(String id) {
					updateClusterState(newClusterStateListener);
				}
			};
		} else {
			nodeListener = null;
		}

		ClusterState cs = null;
		// TODO largeNode
		byte[] stateBuf = zkClient.getNode(zkPath.getStatePath(), nodeListener);
		if (stateBuf != null) {
			logger.info("clusterstate from zk String is  "
					+ new String(stateBuf));
			cs = gson.fromJson(new String(stateBuf), ClusterState.class);
		}

		return cs;
	}

	/**
	 * just for test purpose
	 */
	public void clearState() {
		try {
			logger.info("clear cluster state");
			zkClient.deleteNode(zkPath.getStatePath());
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * a method to re-watch zk node.
	 * 
	 * @param newClusterStateListener
	 */
	private void updateClusterState(
			NewClusterStateListener newClusterStateListener) {
		if (!lifecycle.started()) {
			return;
		}

		try {
			// master node not listener for the state.
			NewClusterStateListener listener = localNode.getRole() == Role.SLAVE ? newClusterStateListener
					: null;

			ClusterState clusterState = retrieveState(listener);
			if (clusterState != null) {
				newClusterStateListener.onNewClusterState(clusterState);
			}
		} catch (ZkClientSessionExpiredException ex) {
			// Ignore session should be restarted
		} catch (Exception ex) {
			logger.error("Error updating cluster state", ex);
		}
	}

	@Override
	public void doStart() {

		logger.info("start " + localNode.getId());
		zkClient.startup();
		if (clearState)
			this.clearState();

		createRootNode();
		createRootLock();

		asyncJoinCluster(true);
	}

	@Override
	public void doStop() {
		logger.info("stop " + localNode.getId());
		zkClient.shutdown();
		localNode.setRole(Role.SLAVE);

		if (currentJoinThread != null) {
			try {
				currentJoinThread.interrupt();
			} catch (Exception e) {
				// ignore
			}
		}
	}

	private void createRootNode() {
		try {
			logger.trace("Creating root nodes in ZooKeeper");
			zkClient.createPersistentNode(zkPath.getNodePath());
		} catch (InterruptedException ex) {
			Thread.currentThread().interrupt();
		}
	}
	
	/**
	 * Lock node for distibute lock operation.
	 */
	private void createRootLock(){
		try {
			logger.trace("Creating root lock in ZooKeeper");
			zkClient.createPersistentNode(zkPath.getLockPath());
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		}
	}

	private void asyncJoinCluster(final boolean initial) {
		// register the cluster
		currentJoinThread = new Thread(new Runnable() {
			public void run() {
				try {
					innerJoinCluster(initial);
				} finally {
					currentJoinThread = null;
				}
			}
		}, "JoinThread");
		currentJoinThread.start();
	}

	/**
	 * join cluster, register self.
	 */
	private void innerJoinCluster(boolean initial) {
		try {
			// if initial, register
			if (!initial || register()) {
				if (localNode.isEnableMaster()) {
					electMaster();
				} else {
					findMaster(initial);
				}
			}
		} catch (InterruptedException ex) {

		}
	}

	/**
	 * first node which occupy the master zk node become master, if we want a
	 * FIX master, then configure a FIX node to be master candidates.
	 */
	private void electMaster() {
		if (lifecycle.stoppedOrClosed()) {
			return;
		}

		try {
			ZkClient.NodeListener nodeListener = new AbstractNodeListener() {
				@Override
				public void onNodeDeleted(String id) {
					handleMasterGone();
				}
			};

			byte[] electedMasterId = zkClient.getOrCreateTransientNode(
					zkPath.getMasterPath(), localNode.getId().getBytes(),
					nodeListener);
			String electedMasterIdStr = new String(electedMasterId);
			if (localNode.getId().equals(electedMasterIdStr)) {
				becomeMaster();
			} else {
				addMaster(electedMasterIdStr);
			}
		} catch (Exception e) {
			logger.error("Couldn't elect master. Restarting discovery.", e);
			restartDiscovery();
		}
	}

	private void becomeMaster() throws InterruptedException {
		logger.info("becoming master " + localNode.getId());
		localNode.setRole(Role.MASTER);
		// init
		initClusterState();

		// listening node join/leave event
		handleUpdateNodeList();
	}

	/**
	 * nodes which is not master candidates, just find master.
	 * 
	 * @param initial
	 * @throws InterruptedException
	 */
	void findMaster(final boolean initial) throws InterruptedException {
		if (lifecycle.stoppedOrClosed()) {
			return;
		}

		ZkClient.NodeListener nodeListener = new AbstractNodeListener() {
			@Override
			public void onNodeCreated(String id) {
				handleMasterAppeared(initial);
			}

			@Override
			public void onNodeDeleted(String id) {
				handleMasterGone();
			}
		};

		byte[] masterId = zkClient
				.getNode(zkPath.getMasterPath(), nodeListener);
		if (masterId == null) {
			/**
			 * if master is null, master is gone or master is not elected yet.
			 * Here master is gone.
			 */
			if (!initial) {
				removeMaster();
			}

		} else {
			addMaster(new String(masterId));
		}
	}

	/**
	 * listening state from master.
	 * 
	 * @param masterId
	 * @throws InterruptedException
	 */
	private void addMaster(String masterId) throws InterruptedException {
		logger.info(localNode.getId() + " add Master: " + masterId);
		localNode.setRole(Role.SLAVE);
		ClusterState state = retrieveState(new NewClusterStateListener() {

			@Override
			public void onNewClusterState(ClusterState state) {
				handleNewClusterStateFromMaster(state);
			}

		});

		/**
		 * There should no change between electMaster and becomeMaster
		 */
		if (state != null && masterId.equals(state.masterNode().getId())) {
			// Check that this state was published by elected master
			handleNewClusterStateFromMaster(state);
		}
	}

	private void handleNewClusterStateFromMaster(final ClusterState clusterState) {
		if (!lifecycle.started()) {
			return;
		}

		for (NewClusterStateListener listener : newClusterStateListeners) {
			try {
				listener.onNewClusterState(clusterState);
			} catch (Exception e) {
				logger.error("handle NewClusterState error ", e);
			}
		}
	}

	private void initClusterState() throws InterruptedException {
		ClusterState cs = retrieveState(null);

		for (InitClusterStateListener listener : initClusterStateListeners) {
			try {
				listener.initClusterState(cs);
			} catch (Exception e) {
				logger.error("handle initClusterState error ", e);
			}
		}
	}

	/**
	 * register the node.
	 * 
	 * @return
	 */
	private boolean register() {
		if (lifecycle.stoppedOrClosed()) {
			return false;
		}

		try {
			String str = gson.toJson(localNode);
			zkClient.setOrCreateTransientNode(nodePath(), str.getBytes());

			return true;
		} catch (Exception e) {
			logger.error("register error " + localNode.getId(), e);
			restartDiscovery();
		}

		return false;
	}

	/**
	 * get and watch the node.
	 */
	private void handleUpdateNodeList() {
		// very small case, not move to started state yet.
		if (!lifecycle.started()) {
			return;
		}

		if (localNode.getRole() != Role.MASTER) {
			logger.info("No longer master - shouldn't monitor node changes");
			return;
		}
		boolean restart = false;
		updateNodeListLock.lock();
		try {
			Set<String> nodes = zkClient.listNodes(zkPath.getNodePath(),
					new ZkClient.NodeListChangedListener() {

						@Override
						public void onNodeListChanged() {
							// for re-watch
							handleUpdateNodeList();
						}
					});

			// ClusterService only need new Node
			// here we just fetch all.
			Collection<Node> nodeList = Lists.newArrayList();
			for (String node : nodes) {
				byte[] data = zkClient.getNode(this.nodePath(node), null);
				if (data != null) {
					Node n = gson.fromJson(new String(data), Node.class);
					nodeList.add(n);
				}
			}

			if (nodeList.isEmpty() || !nodeList.contains(this.localNode))
				nodeList.add(this.localNode);
			logger.info("Updating Nodes " + nodes);
			updateNodeList(nodeList);
		} catch (ZkClientSessionExpiredException ex) {
			restart = true;
		} catch (Exception ex) {
			restart = true;
			logger.error("Couldn't update node list.", ex);
		} finally {
			updateNodeListLock.unlock();
		}
		if (restart) {
			restartDiscovery();
		}
	}

	void updateNodeList(Collection<Node> nodes) {
		for (NodeChangeListener listener : nodeChangeListeners) {
			try {
				listener.onNodeEvent(nodes);
			} catch (Exception e) {
				logger.error("send nodeEvent error", e);
			}
		}
	}

	private void removeMaster() {
		//
	}

	private void handleMasterGone() {
		if (!lifecycle.started()) {
			return;
		}

		logger.info("Master is gone");
		asyncJoinCluster(false);
	}

	private void handleMasterAppeared(boolean initial) {
		if (!lifecycle.started()) {
			return;
		}

		logger.info("New master appeared");
		asyncJoinCluster(initial);
	}

	private void restartDiscovery() {
		if (!lifecycle.started()) {
			return;
		}

		// TODO. sleep ?
		logger.info("Restarting ZK Discovery");
		createRootNode();
		createRootLock();
		localNode.setRole(Role.SLAVE);
		asyncJoinCluster(true);
	}

	private String nodePath(String id) {
		// ip + port
		return zkPath.getNodePath() + "/" + id;
	}

	private String nodePath() {
		// ip + port
		return zkPath.getNodePath() + "/" + localNode.getId();
	}
}
